package com.example.mq.mqserver;

import com.example.mq.common.Consumer;
import com.example.mq.common.MqException;
import com.example.mq.mqserver.core.*;
import com.example.mq.mqserver.datacenter.DiskDataCenter;
import com.example.mq.mqserver.datacenter.MemoryDataCenter;
import lombok.Data;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

// 通过这个类, 来表示 虚拟主机
// 每个虚拟主机下面都管理着自己的 交换机, 队列, 绑定, 消息, 数据
// 同时提供着 api 供上层调用
@Data
public class VirtualHost {
    private String virtualHostName;
    private MemoryDataCenter memoryDataCenter = new MemoryDataCenter();
    private DiskDataCenter diskDataCenter = new DiskDataCenter();
    private Router router = new Router();
    private ConsumerManager consumerManager = new ConsumerManager(this);

    // 创建锁对象
    private final Object exchangeLocker = new Object();
    private final Object queueLocker = new Object();
    private final Object bindingLocker = new Object();

    public VirtualHost(String name) {
        this.virtualHostName = name;

        // 对于 MemoryDataCenter 来说, 不需要额外的初始化操作. 只要对象 new 出来就行
        // 但是, 对于 DiskDataCenter 来说, 需要进行初始化操作: 建库建表和初始化数据
        diskDataCenter.init();

        // 将硬盘数据恢复到内存中
        try {
            memoryDataCenter.recovery(diskDataCenter);
        } catch (IOException | MqException | ClassNotFoundException e) {
            e.printStackTrace();
            System.out.println("[VirtualHost] 恢复内存数据失败!");
        }
    }

    /**
     *  创建交换机
     *  如果交换机不存在, 则创建, 如果存在, 则直接返回
     * @param exchangeName
     * @param exchangeType
     * @param durable
     * @param autoDelete
     * @param arguments
     * @return
     */
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType, boolean durable, boolean autoDelete,
                                   Map<String, Object> arguments) {
        // 把交换机的名字, 添加上虚拟主机作为前缀
        exchangeName = virtualHostName + exchangeName;

        try {
            synchronized (exchangeLocker) {
                // 1.判定该交换机是否已经存在. (通过内存查询)
                Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);

                // 交换机已存在
                if(existsExchange != null) {
                    System.out.println("[VirtualHost - exchangeDeclare] 交换机已经存在! exchangeName=" + exchangeName);
                    return true;
                }

                // 2.创建交换机
                Exchange exchange = new Exchange();
                exchange.setName(exchangeName);
                exchange.setType(exchangeType);
                exchange.setDurable(durable);
                exchange.setAutoDelete(autoDelete);
                exchange.setArguments(arguments);

                // 3.把交换机写入硬盘
                if(durable) {
                    diskDataCenter.insertExchange(exchange);
                }

                // 4.把交换机写入内存
                memoryDataCenter.insertExchange(exchange);

                // 先写硬盘再写内存 (硬盘易出错, 内存就是插入哈希表中一个数据, 不易出错)

                System.out.println("[VirtualHost - exchangeDeclare] 交换机创建成功! exchangeName=" + exchangeName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost - exchangeDeclare] 交换机创建失败! exchangeName=" + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     *  删除交换机
     * @param exchangeName
     * @return
     */
    public boolean exchangeDelete(String exchangeName) {
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                // 1.先找到对应的交换机
                Exchange toDelete = memoryDataCenter.getExchange(exchangeName);
                if(toDelete == null) {
                    throw new MqException("[VirtualHost - deleteExchange] 交换机删除失败, 交换机不存在, 无法删除!");
                }

                // 2.删除硬盘上的数据
                if(toDelete.isDurable()) {
                    diskDataCenter.deleteExchange(exchangeName);
                }

                // 3.删除内存上的数据
                memoryDataCenter.deleteExchange(exchangeName);

                System.out.println("[VirtualHost - deleteExchange] 交换机删除成功! exchangeName=" + exchangeName);
            }
            return true;
        } catch (MqException e) {
            System.out.println("[VirtualHost - deleteExchange] 交换机删除失败! exchangeName=" + exchangeName);
            e.printStackTrace();
            return false;
        }
    }


    /**
     *  创建队列
     * @param queueName
     * @param durable
     * @param exclusive
     * @param autoDelete
     * @param arguments
     * @return
     */
    public boolean queueDeclare(String queueName, boolean durable, boolean exclusive, boolean autoDelete,
                                Map<String, Object> arguments) {
        // 把队列的名字拼接上虚拟主机的名字
        queueName = virtualHostName + queueName;

        try {
            synchronized (queueLocker) {
                // 1.判定队列是否存在
                MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);

                // 队列已存在, 则无需创建, 即返回 true
                if(existsQueue != null) {
                    System.out.println("[VirtualHost - queueDeclare] 队列已经存在! queueName=" + queueName);
                    return true;
                }

                // 2.创建队列对象
                MSGQueue queue = new MSGQueue();
                queue.setName(queueName);
                queue.setDurable(durable);
                queue.setExclusive(exclusive);
                queue.setAutoDelete(autoDelete);
                queue.setArguments(arguments);

                // 3.写硬盘
                if(durable) {
                    diskDataCenter.insertQueue(queue);
                }

                // 4.写内存
                memoryDataCenter.insertQueue(queue);

                System.out.println("[VirtualHost - queueDeclare] 队列创建成功! queueName=" + queueName);
            }
            return true;
        } catch (IOException e) {
            System.out.println("[VirtualHost - queueDeclare] 队列创建失败! queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     *  删除队列
     * @param queueName
     * @return
     */
    public boolean queueDelete(String queueName) {
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLocker) {
                // 1.根据队列的名字, 查询队列对象
                MSGQueue queue = memoryDataCenter.getQueue(queueName);

                // 如果队列为空, 则说明这个删除操作本身就有问题
                if(queue == null) {
                    throw new MqException("[VirtualHost - queueDelete] 队列删除失败, 队列不存在! queueName=" + queueName);
                }

                // 2.删除硬盘数据
                if(queue.isDurable()) {
                    diskDataCenter.deleteQueues(queueName);
                }

                // 3.删除内存数据
                memoryDataCenter.deleteQueue(queueName);

                System.out.println("[VirtualHost - queueDelete] 队列删除成功! queueName=" + queueName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost - queueDelete] 队列删除失败! queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }


    /**
     *  创建绑定
     * @param queueName
     * @param exchangeName
     * @param bindingKey
     * @return
     */
    public boolean queueBind(String queueName, String exchangeName, String bindingKey) {
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;

        try {
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    // 1.判定绑定是否已存在
                    Binding existsBinding = memoryDataCenter.getBinging(exchangeName, queueName);

                    // 如果绑定已存在, 则本次创建绑定就是错误的
                    if(existsBinding != null) {
                        throw new MqException("[VirtualHost - queueBind] 创建绑定失败, 绑定已存在! queueName=" + queueName
                                + ", exchangeName=" + exchangeName);
                    }

                    // 2.验证 bindingKey 是否合法
                    if(!router.checkBindingKey(bindingKey)) {
                        throw new MqException("[VirtualHost - queueBind] 创建绑定失败, bandingKey 非法! bindingKey=" + bindingKey);
                    }

                    // 3.创建 Binding 对象
                    Binding binding = new Binding();
                    binding.setExchangeName(exchangeName);
                    binding.setQueueName(queueName);
                    binding.setBindingKey(bindingKey);

                    // 4.获取一下对应的交换机和队列. 如果交换机或者队列不存在, 这样的绑定也是无法创建的.
                    MSGQueue queue = memoryDataCenter.getQueue(queueName);
                    if(queue == null) {
                        throw new MqException("[VirtualHost - queueBind] 创建绑定失败, 队列不存在! queueName=" + queueName);
                    }
                    Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                    if(exchange == null) {
                        throw new MqException("[VirtualHost - queueBind] 创建绑定失败, 交换机不存在! exchangeName=" + exchangeName);
                    }

                    // 5.写入硬盘
                    if (exchange.isDurable() && queue.isDurable()) {
                        diskDataCenter.insertBinding(binding);
                    }

                    // 6.写入内存
                    memoryDataCenter.insertBinding(binding);

                    System.out.println("[VirtualHost - queueBind] 创建绑定成功! exchangeName=" + exchangeName
                            + ", queueName=" + queueName);
                }
            }

            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost - queueBind] 创建绑定失败! exchangeName=" + exchangeName
                    + ", queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }


    /**
     *  解除绑定
     * @param queueName
     * @param exchangeName
     * @return
     */
    public boolean queueUnbind(String queueName, String exchangeName) {
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;

        try {
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    // 1.获取绑定, 判定该绑定是否存在
                    Binding binding = memoryDataCenter.getBinging(exchangeName, queueName);
                    if(binding == null) {
                        throw new MqException("[VirtualHost - queueUnbind] 解除绑定失败, 绑定不存在! exchangeName=" + exchangeName
                                + ", queueName=" + queueName);
                    }

                    // 2.删除硬盘中数据
                    diskDataCenter.deleteBinding(binding);

                    // 3.删除内存中数据
                    memoryDataCenter.deleteBinding(binding);

                    System.out.println("[VirtualHost - queueUnbind] 解除绑定成功! exchangeName=" + exchangeName
                            + ", queueName=" + queueName);
                }
            }

            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost - queueUnbind] 解除绑定失败! exchangeName=" + exchangeName
                    + ", queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }


    /**
     *  发送消息到指定队列
     * @param exchangeName
     * @param routingKey
     * @param basicProperties
     * @param body
     * @return
     */
    public boolean basicPublish(String exchangeName, String routingKey, BasicProperties basicProperties, byte[] body) {
        try {
            // 1.转换交换机的名字
            exchangeName = virtualHostName + exchangeName;
            // 2.检查 routingKey 是否合法
            if(!router.checkRoutingKey(routingKey)) {
                throw new MqException("[VirtualHost - basicPublish] 发送消息到指定交换机失败, routingKey 非法! routingKey=" + routingKey);
            }
            // 3.查找交换机对象
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if(exchange == null) {
                throw new MqException("[VirtualHost - basicPublish] 发送消息到指定交换机失败, 交换机不存在! exchangeName=" + exchangeName);
            }
            // 4.判定交换机的类型
            if(exchange.getType() == ExchangeType.DIRECT) {
                // 如果以 直接交换机 的方式来转发消息
                // 那么 routingKey 就是队列的名字, 直接把消息转发到对应队列即可 (此时无视绑定关系)
                
                String queueName = virtualHostName + routingKey;
                // 5.构造消息对象
                Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                // 6.查找该队列名对应的对象
                MSGQueue queue = memoryDataCenter.getQueue(queueName);

                // 7.1 队列不存在, 抛出异常
                if(queue == null) {
                    throw new MqException("[VirtualHost - basicPublish] 发送消息到指定交换机失败, 队列不存在! queueName=" + queueName);
                }
                // 7.2 队列存在, 往队列中写入消息
                sendMessage(queue, message);
            } else {
                // 按照 fanout 和 topic 的方式进行消息转发 (前置内容相同,所有放一块写了)

                // 5.找到该交换机关联的所有绑定, 并遍历这些绑定对象
                ConcurrentHashMap<String, Binding> bindingsMap = memoryDataCenter.getBindings(exchangeName);
                for(Map.Entry<String, Binding> entry : bindingsMap.entrySet()) {
                    // 1) 获取绑定对象, 判定对应的队列是否存在
                    Binding binding = entry.getValue();
                    MSGQueue queue = memoryDataCenter.getQueue(binding.getQueueName());
                    // 绑定对应的队列不存在, 则直接跳过
                    if(queue == null) {
                        System.out.println("[VirtualHost - basicPublish] 发送消息到指定交换机失败, 队列不存在! queueName=" + binding.getQueueName());
                        continue;
                    }
                    // 2) 构造消息对象
                    Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                    // 3) 判定这个消息是否能够转发给该队列
                    //    如果是 fanout, 则需要转发给该交换机绑定的所有队列
                    //    如果是 topic, 则需要转发给 bindingKey 和 routingKey 匹配的队列
                    if(!router.route(exchange.getType(), binding, message)) {
                        continue;
                    }
                    // 4) 真正转发消息给队列
                    sendMessage(queue, message);
                }
            }

            System.out.println("[VirtualHost - basicPublish] 消息发送成功!");
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost - basicPublish] 消息发送失败!");
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 发送消息到指定队列
     * @param queue
     * @param message
     */
    private void sendMessage(MSGQueue queue, Message message) throws IOException, MqException, InterruptedException {
        // 此处发送消息, 就是把消息写入 硬盘 和 内存
        int deliverMode = message.getDeliverMode();
        // deliverMode - 持久化 : 1-不持久化, 2-持久化
        if(deliverMode == 2) {
            // 写入硬盘
            diskDataCenter.sendMessage(queue, message);
        }
        // 写入内存
        memoryDataCenter.sendMessage(queue, message);

        // 通知消费者可以消费消息了
        consumerManager.notifyConsume(queue.getName());
    }

    /**
     * 订阅消息
     * 添加一个队列的订阅者, 当队列收到消息之后, 会自动把消息推送给对应的订阅者
     * @param consumerTag 消费者的身份标识
     * @param queueName 消费者订阅哪个队列
     * @param autoAck 队列被消费完之后, 应答的方式. true-自动应答, false-手动应答
     * @param consumer 是一个回调函数, 即消息被处理的方式,具体内容. 此处类型为 函数式接口. 这样后续调用 basicConsume 并且时, 实参可以传递 lambda 表达式
     * @return
     */
    public boolean basicConsume(String consumerTag, String queueName, boolean autoAck, Consumer consumer) {
        // 构造一个 ConsumerEnv 对象, 把这个对象对应的队列找到, 再把这个 ConsumerEnv 对象添加到该队列种
        queueName = virtualHostName + queueName;
        try {
            consumerManager.addConsumer(consumerTag, queueName, autoAck, consumer);
            System.out.println("[VirtualHost - basicConsume] 订阅消息成功! queueName=" + queueName);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost - basicConsume] 订阅消息失败! queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 消费者 手动应答
     * @param queueName
     * @param messageId
     * @return
     */
    public boolean basicAck(String queueName, String messageId) {
        queueName = virtualHostName + queueName;
        try {
            // 1.获取到消息和队列
            Message message = memoryDataCenter.getMessage(messageId);
            if(message == null) {
                throw new MqException("[VirtualHost - basicAck] 消费者手动应答失败, 要确认的消息不存在! messageId=" + messageId);
            }
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if(queue == null) {
                throw new MqException("[VirtualHost - basicAck] 消费者手动应答失败, 要确认的队列不存在! queueName=" + queueName);
            }
            // 2.删除硬盘上的数据
            if(message.getDeliverMode() == 2) {
                diskDataCenter.deleteMessage(queue, message);
            }
            // 3.删除消息中心的数据
            memoryDataCenter.removeMessage(messageId);
            // 4.删除待确认的集合中的数据
            memoryDataCenter.removeMessageWithAck(queueName, messageId);
            System.out.println("[VirtualHost - basicAck] 消费者手动应答成功! queueName=" + queueName
                    + ", messageId=" + messageId);
            return true;
        } catch (MqException | IOException | ClassNotFoundException e) {
            System.out.println("[VirtualHost - basicAck] 消费者手动应答失败! queueName=" + queueName
                    + ", messageId=" + messageId);
            e.printStackTrace();
            return false;
        }
    }
}

